package org.egomsl.mw.segment;

import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.ByteBufferPool;
import org.apache.hadoop.io.ElasticByteBufferPool;
import org.egomsl.mw.CodecUtils;
import org.egomsl.mw.Constants;
import org.egomsl.mw.block.Block;
import org.egomsl.mw.block.BlockImpl;
import org.egomsl.mw.block.BlockScanner;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.EOFException;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Iterator;
import java.util.LinkedList;

public class SegmentBlockReader {
    private static Logger log = LoggerFactory.getLogger(SegmentBlockReader.class);

    private FileSystem fs;

    public SegmentBlockReader(FileSystem fs) {
        this.fs = fs;
    }

    BlockScanner getBlockScanner(Path path) throws IOException {
        FSDataInputStream fsis = fs.open(path);
        int header = fsis.readInt();

        boolean reversed = (header & Constants.FILE_HEADER_ORDER_MASK) > 0;
        if(reversed) {
            return new SegmentFileBlockScanner(fsis);
        }else {
            return new SequentialFileBlockScanner(fsis);
        }
    }


    /**
     * 逆序存储的block读取
     */
    static class SegmentFileBlockScanner implements BlockScanner {

        private Block recordBlock;
        private ByteBufferPool bufferPool;
        private FSDataInputStream fsis;

        public SegmentFileBlockScanner(FSDataInputStream fsis) throws IOException {
            bufferPool = new ElasticByteBufferPool();

            this.fsis = fsis;
        }

        public boolean hasNext() {
            ByteBuffer buffer = null;

            boolean compressed = false;
            int size = 0;

            try {
                if (recordBlock == null) {
                    buffer = ByteBuffer.allocate(4);

                    try {
                        compressed = fsis.read() > 0;
                        size = fsis.readInt();
                    } catch (EOFException eofe) {
                        return false;
                    }

                    if (size > 0) {
                        buffer = fsis.read(bufferPool, size);

                        if (buffer.hasArray()) {
                            if(compressed) {
                                byte[] uncompressedBlock = CodecUtils.uncompress(
                                        buffer.array(), buffer.position(), buffer.remaining());
                                recordBlock = new BlockImpl(ByteBuffer.wrap(uncompressedBlock));
                            }else {
                                recordBlock = new BlockImpl(buffer);
                            }
                        } else {
                            byte[] bytes = new byte[buffer.remaining()];
                            buffer.get(bytes);

                            if(compressed) {
                                bytes = CodecUtils.uncompress(bytes);
                            }
                            recordBlock = new BlockImpl(ByteBuffer.wrap(bytes));
                        }

                        return true;
                    }
                }
            } catch (IOException e) {
                log.error("Error reading storefile.", e);
            }

            return false;
        }

        public Block next() {
            Block block = recordBlock;

            if (recordBlock != null) {
                recordBlock = null;
            }

            return block;
        }


        public void close() {
            if (fsis != null) {
                try {
                    fsis.close();
                } catch (IOException e) {
                    log.warn("error closing input stream.", e);
                }
            }
        }
    }

    /**
     * 顺序存储的block读取
     */
    static class SequentialFileBlockScanner implements BlockScanner {
        private static int MAX_FILE_SIZE = 4 * (1<<20);


        private ByteBuffer buffer;
        private LinkedList<Block> blocks;
        private Iterator<Block> blockIterator;
        private FSDataInputStream fsis;

        public SequentialFileBlockScanner(FSDataInputStream fsis) throws IOException {
            buffer = ByteBuffer.allocate(MAX_FILE_SIZE);
            blocks = new LinkedList<>();
            this.fsis = fsis;

            init();
        }

        private void init() throws IOException {
            boolean compressed;
            int size = fsis.read(buffer.array());

            if(size == MAX_FILE_SIZE) {
                throw new RuntimeException("File size exceeds max size of " + MAX_FILE_SIZE);
            }else if(size <= 0) {
                throw new RuntimeException("Unexpected read size returned " + size);
            }else {
                buffer.position(size);
            }
            buffer.flip();

            while(buffer.remaining() > 0) {
                compressed = buffer.get() > 0;
                size = buffer.getInt();
                if(size == 0) continue;

                if (size > 0 && size <= buffer.remaining()) {
                    if(compressed) {
                        byte[] uncompressedBytes =
                                CodecUtils.uncompress(buffer.array(), buffer.position(), size);
                        blocks.addFirst(new BlockImpl(ByteBuffer.wrap(uncompressedBytes)));
                    }else {
                        blocks.addFirst(new BlockImpl(buffer.array(), buffer.position(), size));
                    }
                    buffer.position(buffer.position() + size);
                }else {
                    throw new RuntimeException("corruption in file. read size "+size+
                            ", but only "+buffer.remaining()+" left");
                }
            }

            blockIterator = blocks.iterator();
        }

        public boolean hasNext() {
            if(blockIterator!= null) return blockIterator.hasNext();
            return false;
        }

        public Block next() {
            if(blockIterator != null) return blockIterator.next();
            return null;
        }

        public void close() {
            if (fsis != null) {
                try {
                    fsis.close();
                } catch (IOException e) {
                    log.warn("error closing input stream.", e);
                }
            }
        }
    }


}
